package org.leiyang.netty.starters;

import com.ibm.etcd.api.Event;
import com.ibm.etcd.client.kv.KvClient;
import com.ibm.etcd.client.kv.WatchUpdate;
import org.leiyang.common.etcd.IConfigCenter;
import org.leiyang.common.holder.OnlineServerHolder;
import org.leiyang.common.redis.ReactiveRedisCache;
import org.leiyang.common.util.AsyncPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.ApplicationArguments;
import org.springframework.boot.ApplicationRunner;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;

import java.util.List;

import static org.leiyang.common.util.Constants.*;

@Component
@Order(value = 1)
public class EtcdStarter implements ApplicationRunner {
    private final Logger log = LoggerFactory.getLogger(getClass());

    private final IConfigCenter configCenter;
    private final OnlineServerHolder onlineServerHolder;
    private final ReactiveRedisCache redisCache;

    public EtcdStarter(IConfigCenter configCenter, OnlineServerHolder onlineServerHolder,
                       ReactiveRedisCache redisCache) {
        this.configCenter = configCenter;
        this.onlineServerHolder = onlineServerHolder;
        this.redisCache = redisCache;
    }

    //监听Msg Server是否在线
    private void monitorMsgServerOnline() {
        //开启相关的watch事件
        AsyncPool.asyncDo(() -> {
            KvClient.WatchIterator watchIterator = configCenter.watchPrefix(online_server_key);
            while (watchIterator.hasNext()) {
                WatchUpdate watchUpdate = watchIterator.next();
                List<Event> eventList = watchUpdate.getEvents();
                eventList.stream()
                        .filter(event -> event.getType().getNumber() == Event.EventType.DELETE.getNumber())
                        .map(event -> event.getKv().getKey().toStringUtf8().replace(online_server_key, ""))
                        .filter(keyIpPort -> keyIpPort.contains("&"))
                        .forEach(keyIpPort -> {
                            String[] ipPortArray = keyIpPort.split("&");
                            String serverIp = ipPortArray[0].replaceAll("_", ".");
                            String port = ipPortArray[1];
                            log.warn("Found MsgServer [{}:{}] has down, will remove its info from cache;", serverIp, port);
                            onlineServerHolder.removeOffLineServer(serverIp, port)
                                    .flatMap(re -> redisCache.deleteKeys(MSG_SER_WEB_INFO.concat(serverIp).concat(":").concat(port),
                                            MSG_SER_RSOCKET_INFO.concat(serverIp).concat(":").concat(port)
                                            )
                                    )
                                    .subscribe(reNum -> log.info("监测到ETCD里的MSGSERVER已失效,开始移除当前MsgServer[{}:{}]记录", serverIp, port));
                        });
            }
        });
    }

    @Override
    public void run(ApplicationArguments args) throws Exception {
        log.info("开始启动监测ETCD里的MsgServer TASK.");
        monitorMsgServerOnline();
    }
}